package com.zeng.ssm.service.support.idempotent;


import com.zeng.ssm.api.support.idempotent.DistributedLockService;
import com.zeng.ssm.dal.mapper.DistributedLockDao;
import com.zeng.ssm.dal.model.DistributedLock;
import com.zeng.ssm.util.DateUtil;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 数据库分布式锁实现
 * <p>
 * 配置内容：
 * # 分布式锁-类型（memcached、redis、database）
 * switch.distributedLock.type=database
 * # 分布式锁-持有者（类型为database时需要，每个服务都不一致）
 * switch.distributedLock.holder=htes_app_1
 *
 * @author zengwanlin
 * @date 2023-02-26T12:02
 */
public class DbDistributedLockServiceImpl implements DistributedLockService, InitializingBean, DisposableBean {

    private final Logger logger = LoggerFactory.getLogger(DbDistributedLockServiceImpl.class);

    private DistributedLockDao distributedLockDao;

    private PlatformTransactionManager transactionManager;

    private String holder;

    private ScheduledThreadPoolExecutor executor;

    public void setDistributedLockDao(DistributedLockDao distributedLockDao) {
        this.distributedLockDao = distributedLockDao;
    }

    public void setTransactionManager(PlatformTransactionManager transactionManager) {
        this.transactionManager = transactionManager;
    }

    public void setHolder(String holder) {
        this.holder = holder;
    }

    @Override
    public boolean setIfAbsent(String key, String value) {
        throw new UnsupportedOperationException("This method is not allowed");
    }

    @Override
    public boolean setIfAbsent(String key, String value, long timeout, TimeUnit unit) {
        return lock(key, value, unit.toSeconds(timeout));
    }

    @Override
    public boolean deleteAnyway(String key) {
        return unlock(key, StringUtils.EMPTY);
    }

    @Override
    public boolean lock(final String key, String value, final long timeout) {
        TransactionTemplate template = new TransactionTemplate(transactionManager);
        template.setPropagationBehavior(TransactionDefinition.PROPAGATION_NOT_SUPPORTED);
        Boolean result = template.execute(new TransactionCallback<Boolean>() {
            @Override
            public Boolean doInTransaction(TransactionStatus transactionStatus) {
                try {
                    DistributedLock record = new DistributedLock();
                    record.setLockKey(key);
                    record.setLockValue(value);
                    record.setLockCount(1);
                    record.setExpireTime(DateUtil.getExpireTime(timeout));
                    record.setHolder(holder);
                    return distributedLockDao.insertSelective(record) == 1;
                } catch (Throwable t) {
                    if (ClassUtils.isAssignableValue(DuplicateKeyException.class, t)) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("DB加锁失败，LOCK_KEY[{}]已存在", key);
                        }
                    } else {
                        if (logger.isDebugEnabled()) {
                            logger.debug("DB加锁失败，" + t.getMessage(), t);
                        }
                    }
                    return false;
                }
            }
        });
        if (logger.isDebugEnabled()) {
            logger.debug("DB加锁，key[{}]，value[{}]，timeout[{}]，结果[{}]", new Object[]{key, value, timeout, result});
        }
        return result;
    }

    @Override
    public boolean unlock(final String key, final String value) {
        TransactionTemplate template = new TransactionTemplate(transactionManager);
        template.setPropagationBehavior(TransactionDefinition.PROPAGATION_NOT_SUPPORTED);
        return template.execute(new TransactionCallback<Boolean>() {
            @Override
            public Boolean doInTransaction(TransactionStatus transactionStatus) {
                try {
                    DistributedLock example = new DistributedLock();
                    example.setLockKey(key);
                    example.setLockValue(value);
                    example.setHolder(holder);
                    int rows = distributedLockDao.deleteByExample(example);
                    if (logger.isDebugEnabled()) {
                        logger.debug("DB解锁，key[{}]，value[{}]，affected rows[{}]", new Object[]{key, value, rows});
                    }
                    return true;
                } catch (Throwable t) {
                    logger.error("DB解锁失败，" + t.getMessage(), t);
                    return false;
                }
            }
        });
    }

    private void initExecutor() {
        executor = new ScheduledThreadPoolExecutor(1, new DefaultThreadFactory("exp-db-lock-%d", false), new ThreadPoolExecutor.DiscardPolicy());
        executor.setMaximumPoolSize(1);
        executor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                List<DistributedLock> lockList = distributedLockDao.selectByHolder(holder);
                if (CollectionUtils.isEmpty(lockList)) {
                    return;
                }
                for (DistributedLock lock : lockList) {
                    if (expired(lock.getExpireTime())) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("DB锁[]已超时，移除", lock.getLockKey());
                        }
                        DistributedLock example = new DistributedLock();
                        example.setLockKey(lock.getLockKey());
                        example.setLockValue(lock.getLockValue());
                        example.setHolder(holder);
                        distributedLockDao.deleteByExample(example);
                    }
                }
            }
        }, 60, 5, TimeUnit.SECONDS);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(distributedLockDao, "distributedLockDao不能为空");
        Assert.notNull(transactionManager, "transactionManager不能为空");
        Assert.hasText(holder, "holder不能为空");
        initExecutor();
    }

    @Override
    public void destroy() throws Exception {
        if (executor != null) {
            if (logger.isDebugEnabled()) {
                logger.debug("停止DB锁超时移除任务线程池");
            }
            executor.shutdown();
        }
    }

    /**
     * 判断是否已过期
     *
     * @param expireTime 过期时间
     * @return
     */
    private boolean expired(String expireTime) {
        return DateUtil.dateTimeToDate(expireTime).before(new Date());
    }

}
